/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package overlayManager;

import java.io.IOException;
import java.util.Set;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;

import org.apache.qpid.client.message.JMSTextMessage;

/**
 * Listener to messages from the controller.
 */
public class ControllerMessageListener implements MessageListener {

    /** Worst state of a broker. */
    private static final int WORST_STATE = 0;

    /** Overlay manager to which the listener belongs. */
    private OverlayManager om;


    /**
     * Constructor.
     *
     * @param om  overlay manager to which the listener belongs
     */
    public ControllerMessageListener(final OverlayManager om) {
        this.om = om;
    }

    /**
     * Invoked when a message from the controller arrives.
     *
     * Analyses the message:
     *
     * If it is a broker failure, new brokers are assigned to all clients who
     * were using the failed broker and a message is sent to them.
     * 
     * If it is a bottleneck, the indicated topic is removed from the indicated
     * broker and changed to a new one.
     *
     * @param msg  received message
     */
    @Override
    public void onMessage(final Message msg) {

        try {
            //get client id from the message
            JMSTextMessage message = (JMSTextMessage) msg;
            String text = message.getText();
            String[] parts = text.split(":");

            if (parts[0].equals("broker failed")) {  //broker failure

                String brokerFailed = parts[1];
                System.out.println("Broker in " + brokerFailed + " failed");
                om.setBrokerState(brokerFailed, WORST_STATE);

                //get the topics published to the failed broker
                Set<String> topicsPubToBrkr
                        = om.getTopicsPubToBroker(brokerFailed);
                for (String propTopicName : topicsPubToBrkr) {
                    //analyse the property name
                    String[] idBr = propTopicName.split(":");
                    String pubId = idBr[0]; //ID of the publisher
                    String pubTopic = idBr[1]; //topic
                    String pubType = idBr[2]; //type (primary or mirror)

                    //change the broker for that topic
                    om.switchBroker(pubTopic, pubId, pubType);
                }

            } else if (parts[0].equals("broker recovered")) { //broker recovery

                String brokerRecovered = parts[1];
                System.out.println("Broker in " + brokerRecovered + " recovered");
                om.setBrokerState(brokerRecovered, Integer.parseInt(parts[2]));

            } else if (parts[0].equals("bottleneck")) { //bottleneck

                String brokerBottleneck = parts[1];
                String pubTopic = parts[2];

                String pubPropName = om.getPubToTopicAndBroker(
                        pubTopic, brokerBottleneck);

                String[] idBr = pubPropName.split(":");
                String pubId = idBr[0]; //ID of the publisher
                String pubType = idBr[2]; //type (primary or mirror)

                //change the broker for that topic
                om.switchBroker(pubTopic, pubId, pubType);

            }

        } catch (JMSException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

}
